package cn.itcast.streaming.task;

import cn.itcast.streaming.bean.CustomRuleAlarmResultModel;
import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.RuleInfoModel;
import cn.itcast.streaming.bean.VehicleInfoModel;
import cn.itcast.streaming.function.asyncio.AsyncHttpQueryFunction;
import cn.itcast.streaming.function.flatmap.CustomRuleAlarmExpressionFunction;
import cn.itcast.streaming.function.flatmap.CustomRuleAlarmVehicleInfoFunction;
import cn.itcast.streaming.function.map.LocationInfoRedisFunction;
import cn.itcast.streaming.function.watermark.CustomRuleAlarmWatermark;
import cn.itcast.streaming.function.window.CustomRuleAlarmWindowFunction;
import cn.itcast.streaming.sink.CustomRuleAlarmMongoSink;
import cn.itcast.streaming.source.MysqlCustomRuleInfoSource;
import cn.itcast.streaming.source.MysqlVehicleInfoFunction;
import cn.itcast.streaming.utils.JsonParsePartUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

//自定义告警实时业务开发
//实时读取kafka数据，将数据过滤转换等操作，然后与车辆基础信息表，地理位置数据进行拉宽操作
//然后自定义告警规则数据进行广播，将广播流数据与应用到事件流之上，将分析后的结果实时写入到MongoDB中
public class CustomRuleAlarmTask extends BaseTask{
    /**
     * 1.创建流处理运行环境
     * 2.接入kafka数据源，实时获取kafka数据
     * 3.将获取到的数据进行过滤到空数据，然后转换为javabean对象
     * 4.过滤掉异常数据，获取正常数据
     * 5.与redis地理位置维度表数据进行关联拉宽地理位置信息，然后根据经纬度生成geohash
     * 6.过滤出来redis拉宽成功的数据
     * 7.过滤出redis拉宽失败的数据
     * 8.与redis拉宽失败的数据，采样异步IO的方式，通过高德地图api获取获取到经纬度对应的地理位置信息，并将获取的结果写入redis中
     * 9.将redis拉宽成功的数据与高德地图阿皮拉宽成功的数据进行合并
     * 10.设置水印操作30s延迟，然后根据vin进行分组，采用90s的滚动窗口，并对窗口自定义函数据
     * 11.查询并加载车辆基础信息表的数据（车辆类型编码表，车辆表，销售记录表，车辆用途表）并进行广播处理
     * 12.将窗口流数据与地理位置广播流数据进行connect操作，然后自定义flatmap函数进行拉宽处理
     * 13.加载自定义告警规则数据，(mysql:监控类型表，监控任务表，规则表），并进行广播
     * 14.将12与自定义告警规则流进行合并，然后应用自定义flatmap函数进行拉宽操作
     *  14.1.判断当前数据是否符合自定义告警规则，如果符合则认为是告警规则数据，反之是非告警数据
     * 15.将最终的结果写入MongoDB中
     *  15.1.同一条数据可能同时满足多个告警监控类型，需要根据监控类型id进行分组统计符合规则的数据
     *  15.2.判断当前告警数据次数是否满足告警阈值，如果满足告警阈值，则写如mongoDB告警状态数据
     *  15.3.有可能告警异常数据会持续一段时间，因此用一辆车的同一个监控类型id在一个告警周期中只需要写入一条数据即可，使用截止时间进行闭合告警周期
     * 16.运行作业
     */
    public static void main(String[] args) {
        //todo 1.创建流处理运行环境
        StreamExecutionEnvironment env = getEnv("CustomRuleAlarmTask");
        env.setParallelism(1);
        //todo 2.接入kafka数据源，实时获取kafka数据
        DataStream<String> kafkaStream = createKafkaStream(SimpleStringSchema.class);
        //todo 3.将获取到的数据进行过滤到空数据，然后转换为javabean对象
        SingleOutputStreamOperator<ItcastDataPartObj> itcastDataPartObjStream = kafkaStream
                .filter(StringUtils::isNotEmpty).map(JsonParsePartUtil::parseJsonToObject);
        //todo 4.过滤掉异常数据，获取正常数据
        SingleOutputStreamOperator<ItcastDataPartObj> srcDataPartObjStream = itcastDataPartObjStream
                .filter(obj -> obj.getErrorData().isEmpty());

        //todo 5.与redis地理位置维度表数据进行关联拉宽地理位置信息，然后根据经纬度生成geohash
        SingleOutputStreamOperator<ItcastDataPartObj> itcastLocationDataObjStream = srcDataPartObjStream
                .map(new LocationInfoRedisFunction());

        //todo 6.过滤出来redis拉宽成功的数据
        SingleOutputStreamOperator<ItcastDataPartObj> itcastWithLocationStream = itcastLocationDataObjStream
                .filter(obj -> StringUtils.isNotEmpty(obj.getProvince()));

        //todo 7.过滤出redis拉宽失败的数据
        SingleOutputStreamOperator<ItcastDataPartObj> itcastNoWithLocationStream = itcastLocationDataObjStream
                .filter(obj -> StringUtils.isEmpty(obj.getProvince()));
        //todo 8.与redis拉宽失败的数据，采样异步IO的方式，通过高德地图api获取获取到经纬度对应的地理位置信息，并将获取的结果写入redis中
        SingleOutputStreamOperator<ItcastDataPartObj> asyncItcastObjWithLocationStream = AsyncDataStream.unorderedWait(
                itcastNoWithLocationStream,   //输入的数据流
                new AsyncHttpQueryFunction(), //异步查询的function对象
                2000,                         //超时时间
                TimeUnit.MICROSECONDS,        //超时时间单位
                10                            //最大的异步并发请求数据（异步的并发线程队列数）
        );
        //todo 9.将redis拉宽成功的数据与高德地图阿皮拉宽成功的数据进行合并
        DataStream<ItcastDataPartObj> itcastDataPartObjWithLocationStream =
                itcastWithLocationStream.union(asyncItcastObjWithLocationStream);

        //todo 10.设置水印操作30s延迟，然后根据vin进行分组，采用90s的滚动窗口，并对窗口自定义函数据
        SingleOutputStreamOperator<ArrayList<ItcastDataPartObj>> itcastWindowDataStream = itcastDataPartObjWithLocationStream
                .assignTimestampsAndWatermarks(new CustomRuleAlarmWatermark())  //添加30s的水印延迟时间
                .keyBy(obj -> obj.getVin())
                .window(TumblingEventTimeWindows.of(Time.seconds(90)))          //90s的时间时间滚动窗口
                .apply(new CustomRuleAlarmWindowFunction());                    //对窗口内的数据排序

        //todo 11.查询并加载车辆基础信息表的数据（车辆类型编码表，车辆表，销售记录表，车辆用途表）并进行广播处理
        DataStream<HashMap<String, VehicleInfoModel>> vehicleInfoModelStream = env
                .addSource(new MysqlVehicleInfoFunction()).broadcast();

        //todo 12.将窗口流数据与地理位置广播流数据进行connect操作，然后自定义flatmap函数进行拉宽处理
        SingleOutputStreamOperator<ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>>> objWithLocationStream =
                itcastWindowDataStream.connect(vehicleInfoModelStream).flatMap(new CustomRuleAlarmVehicleInfoFunction());
        //objWithLocationStream.print("带有车辆基础信息的数据>>>>");

        //todo 13.加载自定义告警规则数据，(mysql:监控类型表，监控任务表，规则表），并进行广播
        DataStream<ArrayList<RuleInfoModel>> ruleInfoStream = env
                .addSource(new MysqlCustomRuleInfoSource()).broadcast();
        ruleInfoStream.print("自定义告警规则数据：");
        //todo 14.将12与自定义告警规则流进行合并，然后应用自定义flatmap函数进行拉宽操作
        //todo 14.1.判断当前数据是否符合自定义告警规则，如果符合则认为是告警规则数据，反之是非告警数据
        SingleOutputStreamOperator<ArrayList<CustomRuleAlarmResultModel>> resultDataStream = objWithLocationStream
                .connect(ruleInfoStream).flatMap(new CustomRuleAlarmExpressionFunction());
        resultDataStream.printToErr("分析后的最终结果》");
        //todo 15.将最终的结果写入MongoDB中
        //todo 15.1.同一条数据可能同时满足多个告警监控类型，需要根据监控类型id进行分组统计符合规则的数据
        resultDataStream.addSink(new CustomRuleAlarmMongoSink());
        //todo 15.2.判断当前告警数据次数是否满足告警阈值，如果满足告警阈值，则写如mongoDB告警状态数据
        //todo 15.3.有可能告警异常数据会持续一段时间，因此用一辆车的同一个监控类型id在一个告警周期中只需要写入一条数据即可，使用截止时间进行闭合告警周期
        //todo 16.运行作业
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
